package com.yyy.modules.core;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.yyy.common.Interceptors.IMsgCallbackHandler;
import com.yyy.entity.Message;
import com.yyy.entity.MessageHandler;
import com.yyy.entity.QueueWrapper;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import javax.annotation.Resource;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;

/**
 * @author yyy
 * @Date 2020-10-18 16:00
 * @Email yangyouyuhd@163.com
 */
@Data
@Slf4j
public class MsgDispatcherService implements ApplicationRunner {

    private MsgFactory factory;
    /**
     * 队列名到队列的映射
     */
    private Map<String, QueueWrapper> mqMapping;

    private Map<String, Integer> polling;

    /**
     * 维护队列名和多个消费者的映射关系
     * 多个消费者监听同一队列时、采用轮询算法
     */
    private Map<String, ArrayList<MessageHandler>> msgHandlerMap;

    @Resource
    private ThreadPoolExecutor executor;

    private Thread pollingThread;

    private AtomicBoolean blocking = new AtomicBoolean(false);

    private List<String> nameList;

    /**
     * 如果用户为某个队列注入了回调处理类、则在此维护它
     */
    private Map<String, IMsgCallbackHandler> callbackHandlerMap;

    @Autowired
    public MsgDispatcherService(MsgFactory factory) {
        this.factory = factory;
        this.mqMapping = factory.getMqMapping();
        this.polling = factory.getPolling();
        this.msgHandlerMap = factory.getMsgHandlerMap();
        this.nameList = factory.getQueueNameList();
        this.callbackHandlerMap = factory.getCallbackHandlerMap();
    }


    /**
     * 消息监听与推送、巨龙线程驻守
     *
     * @param args
     * @throws Exception
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        final Set<Map.Entry<String, QueueWrapper>> set = this.mqMapping.entrySet();
        new Thread(() -> {
            pollingThread = Thread.currentThread();
            while (true) {
                boolean isIdle = true;
                try {
                    for (Map.Entry<String, QueueWrapper> entry : set) {
                        if (!entry.getValue().isEmpty() && entry.getValue().isIdle()) {
                            // 提交给子线程处理任务
                            scanAndSend(entry.getKey(), entry.getValue());
                            isIdle = false;
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                if (isIdle) {
                    // 没有新增的消息事件、进入阻塞状态
                    System.out.println("远古巨龙没发现新的宝藏睡着了");
                    blocking.set(true);
                    LockSupport.park();
                }
            }
        }, "远古巨龙").start();
    }


    /**
     * 推送消息到队列
     *
     * @param name 队列名
     * @param body 消息体
     */
    public void send(String name, Object body) {
        // 将消息推送到指定队列
        Message<String> msg = new Message(JSON.toJSONString(body));
        QueueWrapper queueWrapper = mqMapping.get(name);
        queueWrapper.offer(msg);
        // 巨龙线程是否处于阻塞状态、如果是、唤醒巨龙
        boolean isBlock = blocking.get();
        if (isBlock && blocking.compareAndSet(true, false)) {
            System.out.println(Thread.currentThread().getName() + "唤醒了远古巨龙");
            LockSupport.unpark(pollingThread);
        }
    }

    private void scanAndSend(String name, QueueWrapper queueWrapper) {
        CompletableFuture.runAsync(() -> {
            try {
                if (queueWrapper.tryWork()) {
                    while (!queueWrapper.isEmpty()) {
                        Message message = queueWrapper.remove();
                        Object value = executor(name, message);

                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                queueWrapper.doIdle();
            }

        }, executor);
    }

    private void genericInstanceHandler(Message msg, MessageHandler handler) {
        try {
            Class<?> genericClazz = Class.forName(handler.getGenericClazzName());
            String body = (String) msg.getBody();
            Object genericInstance = JSON.parseObject(body, genericClazz);
            msg.setBody(genericInstance);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    /**
     * 队列为空、直接将消息推送给消费者
     * 1、获取绑定的消费者集合
     * 2、轮询得到一个消费者
     * 3、执行
     *
     * @param name
     * @param msg
     */
    public Object executor(String name, Message msg) {
        MessageHandler handler = this.getPollingHandler(name);
        // 执行
        Method method = handler.getMethod();
        Object instance = handler.getTarget();
        int len = handler.getParamTypeIndexMappingArr().length;
        Object[] args = new Object[len];
        genericInstanceHandler(msg, handler);
        args[handler.getMsgIndex()] = msg;
        Object value = null;
        try {
            value = method.invoke(instance, args);
            // TODO 用户是否对消息手动确认、进行了手动确认执行确认回调、否则add to 队列
            // 成功处理
            IMsgCallbackHandler callbackHandler = callbackHandlerMap.get(name);
            if (ObjectUtil.isNotNull(callbackHandler)) {
                callbackHandler.returnCallback(msg);
            }
            return value;
        } catch (Exception e) {
            // 异常处理
            IMsgCallbackHandler callbackHandler = callbackHandlerMap.get(name);
            if (ObjectUtil.isNotNull(callbackHandler)) {
                callbackHandler.afterCallback(msg, e);
            }
            return value;
        }
    }

    private MessageHandler getPollingHandler(String name) {
        ArrayList<MessageHandler> list = this.msgHandlerMap.get(name);
        int size = list.size();
        Integer exeIndex = this.polling.get(name);
        exeIndex = exeIndex == null ? 0 : exeIndex;
        this.polling.put(name, (exeIndex + 1) % size);
        return list.get(exeIndex);
    }
}

